package com.wuwangfu.state;

import com.wuwangfu.func.QueryStateRichMapFunc;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.QueryableStateOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description：查询状态服务端
 * @Author：jcshen
 * @Date：2023-06-28
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/datastream/fault-tolerance/queryable_state/
 */
public class QueryableKeyedStateServer {

    public static void main(String[] args) throws Exception {
        ParameterTool params = ParameterTool.fromArgs(args);
        Configuration config = params.getConfiguration();
        config.setInteger("rest.port",8081);
        //启用Queryable State服务相关参数
        config.setBoolean(QueryableStateOptions.ENABLE_QUERYABLE_STATE_PROXY_SERVER,true);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);

        //开启checkpoint【检查点，可以将任务计算的中间结果（状态数据保存起来）】
        env.enableCheckpointing(30000);
        //设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));
        //从指定的socket地址和端口创建DataStream
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        //组合kv
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {

                return Tuple2.of(value, 1);
            }
        });
        //按key分组
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndOne.keyBy(t -> t.f0);
        //分完组，形同的key会进入相同的组，每个组都维护自己的状态数据
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = keyed.map(new QueryStateRichMapFunc());

        result.print();

        env.execute();

    }
}
